package com.heima.article.stream;

import com.heima.article.config.KafkaStreamListener;
import com.heima.common.contatns.MQConstants;
import com.heima.model.article.dtos.ArticleVisitStreamMsg;
import com.heima.model.article.dtos.UpdateArticleMsg;
import com.heima.utils.common.JsonUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.checkerframework.checker.units.qual.A;
import org.checkerframework.checker.units.qual.K;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;

/**
 * 统计用户行为的流式处理类
 */
@Component
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    @Override
    public String listenerTopic() {
        return MQConstants.HOT_ARTICLE_INPUT_TOPIC;
    }

    @Override
    public String sendTopic() {
        return MQConstants.HOT_ARTICLE_OUTPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        /**
         * 输入消息格式：
         *   key       value
         *           {"articleId":1001","type","LIKES"}
         *           {"articleId":1001","type","LIKES"}
         *           {"articleId":1002","type","LIKES"}
         *           {"articleId":1002","type","COMMENT"}
         *
         *   定义分组内容格式：
         *       "1001:LIKES"
         *
         *  1) 对value内容进行格式转换
         *   key       value
         *           ["1001:LIKES"]
         *           ["1002:LIKES"]
         *
         * 2) 把value值赋值给key
         *   key                      value
         *  ["1001:LIKES"]         ["1001:LIKES"]
         *  ["1001:LIKES"]         ["1002:LIKES"]
         *
         * 3)对key的内容进行分组统计
         *   key                      value(total)
         *  1001:LIKES               100
         *  1002:LIKES               20
         *  1001:COMMENT             60
         *
         * 4)添加时间窗口设置
         *   key                      value(total)
         *  Window(1001:LIKES)          100
         *  Window（1002:LIKES）        20
         *  Window（1001:COMMENT）     60
         *
         * 目标消息格式：
         *  key       value
         *          {"articleId":1001","like":100,"comment":0....}
         *          {"articleId":1001","like":0,"comment":20....}
         *          {"articleId":1002","like":100,"comment":20....}
         */
        KTable<Windowed<Object>, Long> ktable = stream.flatMapValues(new ValueMapper<String, Iterable<?>>() {
            @Override
            public Iterable<?> apply(String value) { // {"articleId":1001","type","LIKES"}
                UpdateArticleMsg msg = JsonUtils.toBean(value, UpdateArticleMsg.class);
                String msgStr = msg.getArticleId() + ":" + msg.getType().name();
                return Arrays.asList(msgStr);
            }
        }).map(new KeyValueMapper<String, Object, KeyValue<?, ?>>() {

            @Override
            public KeyValue<?, ?> apply(String key, Object value) {
                return new KeyValue<>(value, value);
            }
        }).groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .count(Materialized.as("total"));

        KStream kStream = ktable.toStream().map(new KeyValueMapper<Windowed<Object>, Long, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(Windowed<Object> windowed, Long value) {
                ArticleVisitStreamMsg streamMsg = new ArticleVisitStreamMsg();

                String key = (String)windowed.key();
                if(StringUtils.isNotEmpty(key)){
                    String[] array = key.split(":");
                    //设置文章ID
                    streamMsg.setArticleId(Long.valueOf(array[0]));

                    switch (UpdateArticleMsg.UpdateArticleType.valueOf(array[1])){
                        case LIKES:
                            streamMsg.setLike(value);
                            break;
                        case COMMENT:
                            streamMsg.setComment(value);
                            break;
                        case COLLECTION:
                            streamMsg.setCollect(value);
                            break;
                        case VIEWS:
                            streamMsg.setView(value);
                            break;
                    }

                }
                return new KeyValue<>("",JsonUtils.toString(streamMsg));
            }
        });

        return kStream;
    }
}
